package com.zsy._21生产者与消费者问题;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.sql.Connection;
import java.util.LinkedList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 数据库连接池-等待超时通知范式：加锁-条件循环+超时判断-处理逻辑
 *
 * @author zhangshuaiyin
 * @date 2022/6/11 20:45
 */
@SuppressWarnings("ALL")
public class WaitNotifyTimeout {

    static ConnectionPool pool = new ConnectionPool(10);
    // 保证所有 ConnectionRunner 能够同时开始
    static CountDownLatch start = new CountDownLatch(1);
    // main 线程将会等待所有 ConnectionRunner 结束后才能继续执行
    static CountDownLatch end;

    public static void main(String[] args) throws InterruptedException {
        // 线程数量，可以修改线程数量进行观察
        int threadConunt = 11;
        end = new CountDownLatch(threadConunt);

        int count = 20;
        AtomicInteger got = new AtomicInteger();
        AtomicInteger notGot = new AtomicInteger();

        for (int i = 0; i < threadConunt; i++) {
            new Thread(new ConnectionRunner(count, got, notGot), "ConnectionRunnerThread" + (i + 1))
                    .start();
        }
        start.countDown();
        end.await();
        System.out.println("total invoke: " + (threadConunt * count));
        System.out.println("got connection: " + got);
        System.out.println("not got connection: " + notGot);
    }

    static class ConnectionRunner implements Runnable {
        int count;
        AtomicInteger got;
        AtomicInteger notGot;

        public ConnectionRunner(int count, AtomicInteger got, AtomicInteger notGot) {
            this.count = count;
            this.got = got;
            this.notGot = notGot;
        }

        @Override
        public void run() {
            try {
                start.await();
            } catch (Exception ignored) {
            }
            while (count > 0) {
                try {
                    // 从连接池中获取连接，如果1000ms内无法获取到，将会返回null。
                    // 分别统计连接获取的数量got和未获取到的数量notGot
                    Connection connection = pool.fetchConnection(1000);
                    if (connection == null) {
                        notGot.incrementAndGet();
                    } else {
                        try {
                            ProfilerUtils.begin();
                            connection.createStatement();
                            connection.commit();
                            System.out.println(Thread.currentThread().getName() + "获取到连接, 执行时间为：" + ProfilerUtils.end());
                        } finally {
                            pool.releaseConnection(connection);
                            got.incrementAndGet();
                        }
                    }
                } catch (Exception ignored) {
                } finally {
                    count--;
                }
            }
            end.countDown();
        }
    }

    static class ConnectionPool {
        private LinkedList<Connection> pool = new LinkedList<>();

        public ConnectionPool(int initialSize) {
            if (initialSize > 0) {
                for (int i = 0; i < initialSize; i++) {
                    pool.addLast(ConnectionDriver.createConnection());
                }
            }
        }

        public void releaseConnection(Connection connection) {
            if (connection != null) {
                synchronized (pool) {
                    pool.addLast(connection);
                    // 通知其他消费者有可用连接
                    pool.notifyAll();
                }
            }
        }

        public Connection fetchConnection(long mills) throws InterruptedException {
            synchronized (pool) {
                if (mills <= 0) {
                    while (pool.isEmpty()) {
                        pool.wait();
                    }
                    return pool.removeFirst();
                } else {
                    long future = System.currentTimeMillis() + mills;
                    long remaining = mills;
                    while (pool.isEmpty() && remaining > 0) {
                        pool.wait(remaining);
                        remaining = future - System.currentTimeMillis();
                    }
                    Connection result = null;
                    if (pool.isEmpty() == false) {
                        result = pool.removeFirst();
                    }
                    return result;
                }
            }
        }
    }

    static class ConnectionDriver {
        static class ConnectionHandler implements InvocationHandler {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                if (method.getName().equals("commit")) {
                    TimeUnit.MILLISECONDS.sleep(100);
                }
                return null;
            }
        }

        public static final Connection createConnection() {
            return (Connection) Proxy.newProxyInstance(ConnectionDriver.class.getClassLoader(),
                    new Class[]{Connection.class}, new ConnectionHandler());
        }
    }
}
